package com.sfx.messagequeue.mqServer;

import com.sfx.messagequeue.common.arguments.*;
import com.sfx.messagequeue.common.consumer.Consumer;
import com.sfx.messagequeue.common.exception.MQException;
import com.sfx.messagequeue.common.request.Request;
import com.sfx.messagequeue.common.response.Response;
import com.sfx.messagequeue.common.returns.BasicReturns;
import com.sfx.messagequeue.common.returns.SubscribeReturns;
import com.sfx.messagequeue.common.utils.BinaryPool;
import com.sfx.messagequeue.mqServer.VirtualHost;
import com.sfx.messagequeue.mqServer.core.BasicProperties;

import java.io.*;
import java.net.ServerSocket;
import java.net.Socket;
import java.net.SocketException;
import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

/**
 * Created with IntelliJ IDEA.
 * Description: 消息队列的本体服务器 就是一个TCP的服务器.
 * User: sfx
 * Date: 2023-08-11
 * Time: 18:04
 */
public class BrokerServer {
    // 创建一个虚拟主机
    private VirtualHost virtualHost = new VirtualHost("default");
    // 创建一个socket来处理连接
    private ServerSocket socket;
    // 保存会话信息-->哪一个客户端连接了当前Broker Server
    // key为channelId   val为客户端
    ConcurrentHashMap<String, Socket> sessions = new ConcurrentHashMap<>();

    // 创建一个线程池用于处理客户端的多个请求
    private ExecutorService executorService;
    // 判断服务器是否结束
    private volatile boolean runnable = true;

    public BrokerServer(int port) throws IOException {
        //与端口进行绑定
        socket = new ServerSocket(port);
    }

//    // 启动Broker Server
//    public void start() throws IOException {
//        System.out.println("启动Broker Server");
//        //创建线程池,动态的线程池
//        executorService = Executors.newCachedThreadPool();
//        try{
//            while (runnable) {
//                // 获取一个连接
//                Socket clientSocket = socket.accept();
//                // 交给线程池来处理一个连接-->客户端的请求
//                executorService.submit(() -> {
//                    processConnection(clientSocket);
//                });
//            }
//        }catch (SocketException socketException) {
//            System.out.println("[Broker Server] 连接正常关闭");
//        }
//    }


    public void start() throws IOException {
        System.out.println("[BrokerServer] 启动!");
        executorService = Executors.newCachedThreadPool();
        try {
            while (runnable) {
                Socket clientSocket = socket.accept();
                // 把处理连接的逻辑丢给这个线程池.
                executorService.submit(() -> {
                    processConnection(clientSocket);
                });
            }
        } catch (SocketException e) {
            System.out.println("[BrokerServer] 服务器停止运行!");
            // e.printStackTrace();
        }
    }

    //停止Broker Server
    public void stop() throws IOException {
        // runnable修改为false
        runnable = false;
        // 关闭连接
        socket.close();
        // 关闭线程池
        executorService.shutdownNow();
    }


    // 通过这个方法, 来处理一个客户端的连接.
    // 在这一个连接中, 可能会涉及到多个请求和响应.
    private void processConnection(Socket clientSocket) {
        try (InputStream inputStream = clientSocket.getInputStream();
             OutputStream outputStream = clientSocket.getOutputStream()) {
            // 这里需要按照特定格式来读取并解析. 此时就需要用到 DataInputStream 和 DataOutputStream
            try (DataInputStream dataInputStream = new DataInputStream(inputStream);
                 DataOutputStream dataOutputStream = new DataOutputStream(outputStream)) {

                while (true) {
                    // 1. 读取请求并解析.
                    Request request = readRequest(dataInputStream);
                    // 2. 根据请求计算响应
                    Response response = process(request, clientSocket);
                    // 3. 把响应写回给客户端
                    writeResponse(dataOutputStream, response);
                }
            }
        } catch (EOFException | SocketException e) {
            // 对于这个代码, DataInputStream 如果读到 EOF , 就会抛出一个 EOFException 异常.
            // 需要借助这个异常来结束循环
            System.out.println("[BrokerServer] connection 关闭! 客户端的地址: " + clientSocket.getInetAddress().toString()
                    + ":" + clientSocket.getPort());
        } catch (IOException | ClassNotFoundException | MQException e) {
            System.out.println("[BrokerServer] connection 出现异常!");
            e.printStackTrace();
        } finally {
            try {
                // 当连接处理完了, 就需要记得关闭 socket
                clientSocket.close();
                // 一个 TCP 连接中, 可能包含多个 channel. 需要把当前这个 socket 对应的所有 channel 也顺便清理掉.
                clearClosedSessions(clientSocket);
            } catch (IOException e) {
                e.printStackTrace();
            }
        }
    }

    /**
     * 清理过期的socket
     * 因为一个TCP连接,包含多个Channel,我们要确保socket关闭的时候,channel也从会话中删除掉
     * 另外即使socket异常关闭,channel没有被正常的移除掉,最终还是要通过clearClosedSessions来清理过期的socket
     *
     * @param clientSocket 清理的socket
     */
    private void clearClosedSessions(Socket clientSocket) {
        // 1. 遍历会话,防止迭代器失效,所以不能进行边遍历变删除
        List<String> toDeleteChannelIds = new ArrayList<>();
        for (Map.Entry<String, Socket> entry : sessions.entrySet()) {
            String channelId = entry.getKey();
            toDeleteChannelIds.add(channelId);
        }
        // 2.删除对应的socket
        for (String channelId : toDeleteChannelIds) {
            sessions.remove(channelId);
        }
        System.out.println("[Broker Server] 清理会话完成 channelIds=" + toDeleteChannelIds);
    }

    /**
     * 根据请求计算出响应 ---> 根据request的type,来决定返回什么
     *
     * @param request 请求
     * @param clientSocket 客户端socket
     * @return 返回响应
     */
    private Response process(Request request, Socket clientSocket) throws IOException, ClassNotFoundException, MQException {
        //1. 先对请求进行初步解析成BasicArguments
        BasicArguments basicArguments = (BasicArguments) BinaryPool.toJavaBean(request.getPayload());
        System.out.println("初步解析请求 : [Request] channelId=" + basicArguments.getChannelId() + ", rid ="
                + basicArguments.getRid() + ", length=" + request.getLength() + ", type=" + request.getType());
        // 处理请求之后的成功还是失败
        boolean ok = true;
        //2. 根据请求不同的type类型来处理不同的请求
        int type = request.getType();
        if (type == 0x1) {
            //创建channel
            sessions.put(basicArguments.getChannelId(), clientSocket);
            System.out.println("[Broker Server] 创建channel" + (ok ? "成功" : "失败") + " channelId=" + basicArguments.getChannelId());
        } else if (type == 0x2) {
            //关闭channel
            sessions.remove(basicArguments.getChannelId());
            System.out.println("[Broker Server] 销毁channel" + (ok ? "成功" : "失败") + " channelId=" + basicArguments.getChannelId());
        } else if (type == 0x3) {
            //创建交换机
            ExchangeDeclareArguments arguments = (ExchangeDeclareArguments) basicArguments;
            //执行exchangeDeclare
            ok = virtualHost.exchangeDeclare(arguments.getExchangeName(), arguments.getType(),
                    arguments.isDurable(), arguments.isAutoDelete(), arguments.getArguments());
            System.out.println("[Broker Server] 创建交换机" + (ok ? "成功" : "失败") + " exchangeName=" + arguments.getExchangeName());
        } else if (type == 0x4) {
            //销毁交换机
            ExchangeDeleteArguments arguments = (ExchangeDeleteArguments) basicArguments;
            //执行exchangeDelete
            ok = virtualHost.exchangeDelete(arguments.getExchangeName());
            System.out.println("[Broker Server] 销毁交换机" + (ok ? "成功" : "失败") + " exchangeName=" + arguments.getExchangeName());
        } else if (type == 0x5) {
            //创建queue
            QueueDeclareArguments arguments = (QueueDeclareArguments) basicArguments;
            ok = virtualHost.queueDeclare(arguments.getQueueName(), arguments.isDurable(), arguments.isAutoDelete()
                    , arguments.isExclusive(), arguments.getArguments());
            System.out.println("[Broker Server] 创建队列" + (ok ? "成功" : "失败") + " queueName=" + arguments.getQueueName());
        } else if (type == 0x6) {
            //销毁queue
            QueueDeleteArguments arguments = (QueueDeleteArguments) basicArguments;
            ok = virtualHost.queueDelete(arguments.getQueueName());
            System.out.println("[Broker Server] 销毁队列" + (ok ? "成功" : "失败") + " queueName=" + arguments.getQueueName());
        } else if (type == 0x7) {
            //创建binding
            QueueBindArguments arguments = (QueueBindArguments) basicArguments;
            ok = virtualHost.queueBind(arguments.getExchangeName(), arguments.getQueueName(), arguments.getBindingKey());
            System.out.println("[Broker Server] 创建Binding" + (ok ? "成功" : "失败") + " queueName=" + arguments.getQueueName()
                    + " exchangeName=" + arguments.getExchangeName());
        } else if (type == 0x8) {
            //销毁binding
            QueueUnBindArguments arguments = (QueueUnBindArguments) basicArguments;
            ok = virtualHost.queueUnbind(arguments.getQueueName(), arguments.getExchangeName());
            System.out.println("[Broker Server] 销毁Binding" + (ok ? "成功" : "失败") + " queueName=" + arguments.getQueueName()
                    + " exchangeName=" + arguments.getExchangeName());
        } else if (type == 0x9) {
            //发送Message
            BasicPublishArguments arguments = (BasicPublishArguments) basicArguments;
            ok = virtualHost.basicPublish(arguments.getExchangeName(), arguments.getBasicProperties(),
                    arguments.getRoutingKey(), arguments.getBody());
            System.out.println("[Broker Server] 发送Message" + (ok ? "成功" : "失败") + " routingKey=" + arguments.getRoutingKey()
                    + " exchangeName=" + arguments.getExchangeName());
        } else if (type == 0xa) {
            //订阅Message
            BasicConsumeArguments arguments = (BasicConsumeArguments) basicArguments;
            ok = virtualHost.basicConsume(arguments.getConsumerTag(), arguments.getQueueName(), arguments.isAutoAck(),
                    new Consumer() {
                        // 这个是固定的回调,服务器接收到订阅者的订阅,服务器收到消息之后,就会把消息推送给消费者
                        // 服务器只需把消息返回给服务器即可,不需要管他做什么
                        @Override
                        public void handleDelivery(String consumerTag, BasicProperties basicProperties, byte[] body) throws MQException, IOException {
                            //我们要把消息推送给消费者，首先得需要知道消费者的客户端是谁
                            //1. 找到消费者的客户端,这里的get要填入参数是channelId，用于标识这个channel-》连接，这里的channelId就是consumerTag
                            Socket clientSocket = sessions.get(consumerTag);
                            //2. 客户端是否存在，不存在直接报错
                            if (clientSocket == null || clientSocket.isClosed()) {
                                throw new MQException("[Broker Server] 客户端不存在或者已经关闭 ip :" + clientSocket.getInetAddress().toString() + ":" + clientSocket.getPort());
                            }
                            //3. 构造响应
                            // 3.1 响应中的payLoad其实就是我们要给消费者推送的消息，之前我们已经封装好了一个推送给消费者消息的参数 subscribeReturns
                            // 3.2 构造subscribeReturns
                            SubscribeReturns subscribeReturns = new SubscribeReturns();
                            subscribeReturns.setBody(body);
                            subscribeReturns.setConsumerTag(consumerTag);
                            // 由于服务器收到生产者给的消息之后，就会推送给消费者，没有请求，只有响应
                            subscribeReturns.setRid("");
                            subscribeReturns.setChannelId(consumerTag);
                            subscribeReturns.setBasicProperties(basicProperties);
                            // 将subscribeReturns转化为一个字节数组写入到响应中
                            byte[] payLoad = BinaryPool.toByteArray(subscribeReturns);
                            Response response = new Response();
                            response.setLength(payLoad.length);
                            response.setPayload(payLoad);
                            // 服务器给客户端推送的消息->被订阅的消息, 是响应独有的.
                            response.setType(0xc);
                            //4. 把响应写入到客户端的socket中
                            // 4.1 得到输出流对象->客户端socket的输出流对象
                            DataOutputStream dataOutputStream = new DataOutputStream(clientSocket.getOutputStream());
                            // 4.2 将响应写入到该流对象
                            writeResponse(dataOutputStream, response);
                        }
                    });
            System.out.println("[Broker Server] 消费者订阅" + (ok ? "成功" : "失败") + "consumerTag=" + arguments.getConsumerTag());
        } else if (type == 0xb) {
            //调用basicAck来确认消息
            BasicAckArguments arguments = (BasicAckArguments) basicArguments;
            ok = virtualHost.basicAck(arguments.getQueueName(), arguments.getMessageId());
            System.out.println("[Broker Server] basicAck确认" + (ok ? "成功" : "失败") + "messageId=" + arguments.getMessageId());
        } else {
            throw new MQException("[Broker Server] 类型错误 type=" + request.getType());
        }
        //封装响应返回
        BasicReturns returns = new BasicReturns();
        returns.setOk(ok);
        returns.setRid(basicArguments.getRid());
        returns.setChannelId(basicArguments.getChannelId());
        byte[] payLoad = BinaryPool.toByteArray(returns);
        Response response = new Response();
        response.setType(request.getType());
        response.setLength(payLoad.length);
        response.setPayload(payLoad);
        return response;
    }



    /**
     * 读取请求
     *
     * @param dataInputStream 进行读
     * @return 返回请求
     */
    private Request readRequest(DataInputStream dataInputStream) throws IOException {
        // 按照我们约定的请求格式进行读取
        Request request = new Request();
        // 1. 先读取类型type->4byte
        request.setType(dataInputStream.readInt());
        // 2. 在读取长度->4byte
        int exceptedLength = dataInputStream.readInt();
        request.setLength(exceptedLength);

        // 3. 根据长度读取payLoad
        byte[] payLoad = new byte[exceptedLength];
        int actualLength = dataInputStream.read(payLoad);
        if (actualLength != exceptedLength) {
            throw new IOException("[Broker Server] 读取客户端请求长度不一致-请求出错");
        }
        request.setPayload(payLoad);
        return request;
    }

    /**
     * 写入响应
     *
     * @param dataOutputStream 输出流 - 写
     * @param response         响应
     */
    private void writeResponse(DataOutputStream dataOutputStream, Response response) throws IOException {
        int type = response.getType();
        // 写入类型
        dataOutputStream.writeInt(type);
        // 写入长度
        int length = response.getLength();
        dataOutputStream.writeInt(length);
        // 写入payLoad
        byte[] payload = response.getPayload();
        if (length != payload.length) {
            throw new IOException("[Broker Server] 写入响应时,读取响应长度与payLoad长度不一致-响应出错");
        }
        dataOutputStream.write(payload);
        // 刷新缓冲区
        dataOutputStream.flush();
    }
}